package com.apobates.jforum2.troll.notices.biz.impl;

import com.apobates.jforum2.troll.notices.ForumLetterReplica;
import com.apobates.jforum2.troll.notices.biz.InboxService;
import com.apobates.jforum2.troll.notices.biz.dao.InboxDao;
import com.apobates.jforum2.troll.notices.entity.ForumLetter;
import com.apobates.jforum2.troll.notices.entity.ForumLetterStatus;
import com.apobates.jforum2.troll.notices.entity.ForumLetterTypeEnum;
import com.apobates.jforum2.troll.notices.entity.Inbox;
import com.apobates.jforum2.troll.utils.core.DateTimeUtils;
import com.apobates.jforum2.troll.utils.persistence.Page;
import com.apobates.jforum2.troll.utils.persistence.Pageable;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 *
 * @author xiaofanku
 * @since 20200510
 */
@Service
public class InboxServiceImpl implements InboxService {
    @Autowired
    private InboxDao inboxDao;

    @Override
    public Page<ForumLetterReplica> getInBox(long memberId, Pageable pageable) {
        Page<ForumLetter> rs = inboxDao.findAllByReceiver(memberId, pageable);

        final Stream<ForumLetterReplica> result = associationLetterStatusAsync(rs.getResult().collect(Collectors.toList()));
        return new Page<ForumLetterReplica>() {
            @Override
            public long getTotalElements() {
                return rs.getTotalElements();
            }

            @Override
            public Stream<ForumLetterReplica> getResult() {
                return result;
            }
        };
    }

    @Override
    public Page<ForumLetter> getInBoxGroupSender(long memberId, Pageable pageable) {
        return inboxDao.findAllByReceiverGroupSender(memberId, pageable);
    }

    @Override
    public Page<ForumLetterReplica> getInBox(long memberId, ForumLetterTypeEnum label, Pageable pageable) {
        Page<ForumLetter> rs = inboxDao.findAllByReceiverAndType(memberId, label, pageable);

        final Stream<ForumLetterReplica> result = associationLetterStatusAsync(rs.getResult().collect(Collectors.toList()));
        return new Page<ForumLetterReplica>() {
            @Override
            public long getTotalElements() {
                return rs.getTotalElements();
            }

            @Override
            public Stream<ForumLetterReplica> getResult() {
                return result;
            }
        };
    }

    @Override
    public Stream<ForumLetter> getReadableMessages(long memberId, int size) {
        return inboxDao.findAllForUnReadable(memberId, size);
    }

    @Override
    public Stream<ForumLetter> get(long sender, long memberId) {
        return inboxDao.findAll(sender, memberId);
    }

    @Override
    public Stream<ForumLetter> get(long sender, long memberId, int prevUnixStamp) {
        LocalDateTime startDateTime = DateTimeUtils.getDateTimeByUnixTimestamp(prevUnixStamp);
        return inboxDao.findAll(sender, memberId, startDateTime);
    }

    @Override
    public Optional<Boolean> readed(long memberId, long sender) {
        return inboxDao.editReadabled(memberId, sender);
    }

    @Override
    public int remove(long memberId, List<Long> idList) {
        if (null == idList || idList.isEmpty()) {
            return 0;
        }
        return inboxDao.editDeleted(memberId, idList);
    }

    @Override
    public int readed(long memberId, List<Long> idList) {
        if (null == idList || idList.isEmpty()) {
            return 0;
        }
        return inboxDao.editReadabled(memberId, idList);
    }

    @Override
    public int readed(long memberId) {
        return inboxDao.editReadabled(memberId);
    }

    @Override
    public Optional<Boolean> compositeReaded(long memberId, long othMemberId, int direction) {
        if (direction == 1) { //将两个会员之间(指定发件人发来)的未读消息标为已读
            return readed(memberId, othMemberId);
        }
        if (othMemberId == -1 && direction == 2) { //将会员的未读消息标记为已读
            return Optional.of(readed(memberId)>0);
        }
        return Optional.empty();
    }

    @Override
    public long countForMemberMessages(long memberId) {
        return inboxDao.countForUnReadable(memberId);
    }

    @Override
    public Map<Long, Long> groupForMemberMessages(long memberId, Set<Long> senderIdSet) {
        if (senderIdSet.isEmpty()) {
            return Collections.emptyMap();
        }
        return inboxDao.groupForUnReadable(memberId, senderIdSet);
    }

    @Override
    public long countForMemberMessages(long memberId, long sender) {
        return inboxDao.countForUnReadable(memberId, sender);
    }

    //[异步]级联加载信件的状态
    private Stream<ForumLetterReplica> associationLetterStatusAsync(final List<ForumLetter> data) {
        if (null == data || data.isEmpty()) {
            return Stream.empty();
        }
        final Map<Long, Inbox> boxes = CompletableFuture.supplyAsync(()->data.stream().map(ForumLetter::getId).collect(Collectors.toSet()))
                .thenCompose(letterIdSet-> CompletableFuture.supplyAsync(()->inboxDao.findAllByLetter(letterIdSet).collect(Collectors.toMap(Inbox::getLetter, Function.identity())))
                        .completeOnTimeout(Collections.emptyMap(), 1, TimeUnit.SECONDS))
                .orTimeout(1, TimeUnit.SECONDS)
                .join();
        Consumer<ForumLetterReplica> action = fl->{
            Inbox ti = boxes.get(fl.getId());
            fl.setStatus(new ForumLetterStatus(ti.getId(), ti.isReadable(), ti.isReply(), ti.isUsable()));
        };
        return data.parallelStream().map(ForumLetterReplica::new).peek(action);
    }
}
